package com.spzx.common.rabbit.config;

import com.alibaba.fastjson.JSON;
import com.spzx.common.rabbit.entity.GuiguCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

//    @Override
//    public void onApplicationEvent(ApplicationReadyEvent event) {
//        this.setupCallbacks();
//    }


    /*    * 只确认消息是否正确到达 Exchange 中,成功与否都会回调
     *
     * @param correlation 相关数据  非消息本身业务数据
     * @param ack             应答结果
     * @param reason           如果发送消息到交换器失败，错误原因
     */
//    private void setupCallbacks(){
//      rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
//            @Override
//            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
//                log.info("correlationData="+correlationData);
//                if (ack){
//                    //消息到交换器成功
//                    log.info("表示交换机[收到]消息了");
//                }else {
//                    log.info("表示交换机[没收到消息]");
//                    log.info("cause="+cause);
//                }
//            }
//        });
//
//        *//**
//         * 消息没有正确到达队列时触发回调，如果正确到达队列不执行
//         *//*
//        this.rabbitTemplate.setReturnsCallback(returned -> {
//            log.error("Returned: " + returned.getMessage() + "\nreplyCode: " + returned.getReplyCode()
//                    + "\nreplyText: " + returned.getReplyText() + "\nexchange/rk: "
//                    + returned.getExchange() + "/" + returned.getRoutingKey());
//        });
//    }


    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallbacks();
    }

/*    private void setupCallbacks() {
        this.rabbitTemplate.setConfirmCallback((correlationData, ack, reason) -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (ack) {
                //消息到交换器成功
                log.info("消息发送到Exchange成功：{}", correlationData);
            } else {
                log.error("消息发送到Exchange失败：{}", reason);
                this.retrySendMsg(correlationData);
            }
        });

        // 消息没有正确到达队列时触发回调，如果正确到达队列不执行
        this.rabbitTemplate.setReturnsCallback(returned -> {
            log.error("Returned: " + returned.getMessage() + "\nreplyCode: " + returned.getReplyCode()
                    + "\nreplyText: " + returned.getReplyText() + "\nexchange/rk: "
                    + returned.getExchange() + "/" + returned.getRoutingKey());

            //当路由队列失败 也需要重发
            String redisKey = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            String correlationDataStr = (String) redisTemplate.opsForValue().get(redisKey);
            GuiguCorrelationData guiguCorrelationData = JSON.parseObject(correlationDataStr, GuiguCorrelationData.class);
            //2.调用消息重发方法
            retrySendMsg(guiguCorrelationData);
        });
    }

    private void retrySendMsg(CorrelationData correlationData) {
        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData) correlationData;

        int retryCount = guiguCorrelationData.getRetryCount();
        if (retryCount >= 3) {
            log.error("生产者超过最大重试次数，将失败的消息存入数据库用由人工处理；给管理员发送邮件；给管理员发送短信；");
            return;
        }
        //重发消息
        rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(), guiguCorrelationData.getRoutingKey(), guiguCorrelationData.getMessage(), guiguCorrelationData);

        //重发次数+1
        retryCount += 1;
        guiguCorrelationData.setRetryCount(retryCount);
        redisTemplate.opsForValue().set(guiguCorrelationData.getId(), JSON.toJSONString(guiguCorrelationData), 10, TimeUnit.MINUTES);
        log.info("进行消息重发!");
    }*/














    private void setupCallbacks() {
        this.rabbitTemplate.setConfirmCallback((correlationData, ack, reason) -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (ack) {
                //消息到交换器成功
                log.info("消息发送到Exchange成功：{}", correlationData);
            } else {
                log.error("消息发送到Exchange失败：{}", reason);
                this.retrySendMsg(correlationData);
            }
        });

        // 消息没有正确到达队列时触发回调，如果正确到达队列不执行
        this.rabbitTemplate.setReturnsCallback(returned -> {
            log.error("Returned: " + returned.getMessage() + "\nreplyCode: " + returned.getReplyCode()
                    + "\nreplyText: " + returned.getReplyText() + "\nexchange/rk: "
                    + returned.getExchange() + "/" + returned.getRoutingKey());

            //当路由队列失败 也需要重发
            String redisKey = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            String correlationDataStr = (String) redisTemplate.opsForValue().get(redisKey);
            GuiguCorrelationData guiguCorrelationData = JSON.parseObject(correlationDataStr, GuiguCorrelationData.class);

            //方案一： 是延迟消息就不去重试；不是延迟消息才会重试
            if (guiguCorrelationData.isDelay()){
                return;
            }
            //2.调用消息重发方法
            retrySendMsg(guiguCorrelationData);
        });
    }

    private void retrySendMsg(CorrelationData correlationData) {
        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData) correlationData;

        int retryCount = guiguCorrelationData.getRetryCount();
        if (retryCount >= 3) {
            log.error("生产者超过最大重试次数，将失败的消息存入数据库用由人工处理；给管理员发送邮件；给管理员发送短信；");
            return;
        }
        //重发消息
        rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(), guiguCorrelationData.getRoutingKey(), guiguCorrelationData.getMessage(), guiguCorrelationData);

        //重发次数+1
        retryCount += 1;
        guiguCorrelationData.setRetryCount(retryCount);
        redisTemplate.opsForValue().set(guiguCorrelationData.getId(), JSON.toJSONString(guiguCorrelationData), 10, TimeUnit.MINUTES);
        log.info("进行消息重发!");

        //方案二: 延迟消息和非延迟消息发送的方式不同。单独处理。
        if(guiguCorrelationData.isDelay()){
            //发送消息同时，把关联数据也发送。回退时需要把关联数据也进行回退，根据关联数据就可以进行消息重复。
            rabbitTemplate.convertAndSend(
                    guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    messageObj -> {
                        messageObj.getMessageProperties().setDelay(guiguCorrelationData.getDelayTime() * 1000); //单位毫秒
                        return messageObj;
                    },
                    correlationData);
        }else{
            //重发
            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    guiguCorrelationData);
        }

        //覆盖缓存数据。
        redisTemplate.opsForValue().set(correlationData.getId(),
                JSON.toJSONString(correlationData),5, TimeUnit.MINUTES);
    }
}
